package com.surfilter.massdata.spark.task.daystati;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;

import com.act.sparkanalyz.service.impl.SparkService.OutQueueEntity;
import com.act.sparkanalyz.task.ISparkTask;
import com.surfilter.massdata.spark.bean.SAN011;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.DateUtil;

/**
 * 访问省内网站情况
 * 本省网民访问省内网站情况（本省网民是指CIP在BR2002中存在，省内网站是指DIP在BR2002中存在）
 */
public class WebSiteVisitsProvinceTask_New implements ISparkTask{
	
	private static final Log log = LogFactory.getLog(WebSiteVisitsProvinceTask_New.class);
	private static final long serialVersionUID = 1L;
	private String outName;
	
	@Override
	public List<OutQueueEntity> execute(Map<String, DataFrame> dataFrames,Map<String, String> commandMap) {
		long start=System.currentTimeMillis();
		log.info("==============WebSiteVisitsProvinceTask begin====================");
		List<OutQueueEntity> outList = new ArrayList<OutQueueEntity>();
		try{
			DataFrame dataSource = dataFrames.get("dwb_domain_cip_dip");
			
	        String dayStr = commandMap.get("-d");
	        Date date = DateUtil.getExecDate(DateUtil.getYesterday(),dayStr);
	        
	        DataFrame result = createRdd(date,dataSource);
			outList.add(new OutQueueEntity(outName, result));
			
			CommonUtils.deleteTaskTableData("SAN011", date, "WEBSITE_COUNT_AREA", 0, "day");
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
		long end=System.currentTimeMillis();
		double min=(end-start)*1.0/(1000*60);
		System.out.println("WebSiteVisitsProvinceTask_New:exectime: "+min+" min............");
		return outList;
	}

	
	private static DataFrame createRdd(final Date date,DataFrame dataSource) {
		
		String sql = "select sum(visit_count) as count,dip_city from dwb_domain_cip_dip_temp group by dip_city";
		DataFrame df = dataSource.sqlContext().sql(sql);
		
		JavaRDD<Map<String ,String>> rdd = df.toJavaRDD().map(new Function<Row, Map<String,String>>() {
			private static final long serialVersionUID = 1L;

			@Override
			public Map<String, String> call(Row row) throws Exception {
				Map<String, String> map = new HashMap<String, String>();
				map.put("count", row.getAs("count").toString());
				map.put("dip_city", row.getAs("dip_city").toString());
				return map;
			}
		});
		
		JavaRDD<SAN011> javaRdd_SAN011 = rdd.mapPartitions(new FlatMapFunction<Iterator<Map<String,String>>, SAN011>() {
			private static final long serialVersionUID = 1L;

			@Override
			public Iterable<SAN011> call(Iterator<Map<String, String>> it) throws Exception {
				List<SAN011> list = new ArrayList<SAN011>();
				try{
					while(it.hasNext()){
						Map<String,String> mapSeq = it.next();
						SAN011 sa = new SAN011();
						String city = String.valueOf(mapSeq.get("dip_city"));
						long count = Long.parseLong(String.valueOf(mapSeq.get("count")));
			    		sa.setYear(DateUtil.getCurrentYear(date));
			    		sa.setHalf_year(DateUtil.getHalfYear(date));
			    		sa.setQuarter(DateUtil.getQuarter(date));
			    		sa.setMonth(DateUtil.getCurrentMonth(date));
			    		sa.setWeek(DateUtil.getCurrentWeek(date));
			    		sa.setDay(DateUtil.getCurrentDay(date));
			    		sa.setBuss_type("WEBSITE_COUNT_AREA");
			    		sa.setBuss_value(city);
			    		sa.setCount_value(count);
			    		sa.setSta_range(0);
			    		list.add(sa);
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return list;
			}
		});
		DataFrame resultDf = dataSource.sqlContext().createDataFrame(javaRdd_SAN011, SAN011.class);
		return resultDf;
	}
}
